#ifndef __M_SERVER_H__
#define __M_SERVER_H__
#include <vector>
#include <cassert>
#include <cstdint>
#include <algorithm>
#include <string>
#include <map>
#include <iostream>
#include <functional>
#include <unordered_map>
#include <mutex>
#include <condition_variable>
#include <cstring>
#include <memory>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <signal.h>
#include <thread>
#include <unistd.h>
#include <fcntl.h>
#include "common.hpp"

#define DEFAULT_BUFFER_SIZE 1024

class Any
{
    private:
        class holder
        {
            public:
                virtual ~holder() {}
                virtual const std::type_info &type() = 0;
                virtual holder *clone() = 0;
        };
        template<class T>
        class placeholder : public holder
        {
            public:
                placeholder(const T &val): _val(val) {}
                // 获取子类对象保存的数据类型
                virtual const std::type_info &type() { return typeid(T); }
                // 根据当前子类保存的数据，克隆一份新的子类对象
                virtual holder *clone() { return new placeholder(_val); }
            public:
                T _val;
        };
        holder *_content;
    public:
        Any(): _content(nullptr) {}
        template<class T>
        Any(const T& val): _content(new placeholder<T>(val)) {}
        Any(const Any& other): _content(other._content ? other._content->clone() : nullptr) {}
        ~Any() { delete _content; }
        
        Any &swap(Any& other)
        {
            std::swap(_content, other._content);
            return *this;
        }
        template<class T>
        T *get()        // 返回当前保存变量的指针
        {
            // 保存的数据类型，必须和要获取的数据类型一直
            assert(_content->type() == typeid(T));
            return &((placeholder<T>*)_content)->_val;
        }
        template<class T>
        Any &operator=(const T &val)       // 赋值运算符重载
        {
            Any(val).swap(*this);
            return *this;
        }
        Any &operator=(const Any &other)
        {
            Any(other).swap(*this);
            return *this;
        }
};

class Buffer
{
    private:
        std::vector<char> _buffer;  // 缓冲区
        uint64_t _read_idx;         // 读偏移
        uint64_t _write_idx;        // 写偏移
    public:
        Buffer(): _read_idx(0), _write_idx(0), _buffer(DEFAULT_BUFFER_SIZE) {}
        char *Begin() { return &*(_buffer.begin()); }
        // 获取读偏移位置
        char *GetReadPosition() { return Begin() + _read_idx; }
        // 获取写偏移位置
        char * GetWritePosition() { return Begin() + _write_idx; }
        // 获取起始空间大小
        uint64_t HeadIdleSize() { return _read_idx; }
        // 获取末尾空间大小
        uint64_t TailIdleSize() { return _buffer.size() - _write_idx; }
        // 获取可读数据大小
        uint64_t ReadAbleSize() { return _write_idx - _read_idx; }
        // 向后移动读偏移
        void MoveReadOffset(uint64_t len)
        {
            if(len == 0) return;
            assert(len <= ReadAbleSize());
            _read_idx += len; 
        }
        // 向后移动写偏移
        void MoveWriteOffset(uint64_t len)
        {
            if(len == 0) return;
            assert(len <= TailIdleSize());
            _write_idx += len;
        }
        // 确保可写空间充足
        void EnsureWriteSpace(uint64_t len)
        {
            // 空间足够，无需调整
            if(len <= TailIdleSize())
            {
                return;
            }
            // 前置空间+末尾空间大小足够，将偏移量清空(拷贝数据至初始点)
            if(len <= TailIdleSize() + HeadIdleSize())
            {
                int rbs = ReadAbleSize();
                std::copy(GetReadPosition(), GetReadPosition() + rbs, Begin());
                _read_idx = 0;
                _write_idx = rbs;
            }
            // 总空间大小不足，直接扩容
            else
            {
                _buffer.resize(_write_idx + len);
            }
        }
        // 写入数据
        void Write(const void *data, uint64_t len)
        {
            if(len == 0) return;
            EnsureWriteSpace(len);
            const char *d = (char*)data;
            std::copy(d, d+len, GetWritePosition());
        }
        // 写入数据并移动写偏移
        void WriteAndMove(const void *data, uint64_t len)
        {
            Write(data, len);
            MoveWriteOffset(len);
        }
        // 传入数据类型为string
        void WriteFromString(const std::string &data)
        {
            Write(data.c_str(), data.size());
        }
        void WriteFromStringAndMove(const std::string &data)
        {
            WriteAndMove(data.c_str(), data.size());
        }
        // 传入数据类型为缓冲区(从其他缓冲区读入数据并写入当前缓冲区)
        void WriteFromBuffer(Buffer &data)
        {
            Write(data.GetReadPosition(), data.ReadAbleSize());
        }
        void WriteFromBufferAndMove(Buffer &data)
        {
            WriteAndMove(data.GetReadPosition(), data.ReadAbleSize());
        }
        // 读取数据
        void Read(void *buf, uint64_t len)
        {
            assert(len <= ReadAbleSize());
            std::copy(GetReadPosition(), GetReadPosition()+len, (char*)buf);
        }
        void ReadAndMove(void *buf, uint64_t len)
        {
            Read(buf, len);
            MoveReadOffset(len);
        }
        std::string ReadAsString(uint64_t len)
        {
            std::string str;
            str.resize(len);
            Read(&str[0], len);
            return str;
        }
        std::string ReadAsStringAndMove(uint64_t len)
        {
            std::string str;
            str.resize(len);
            ReadAndMove(&str[0], len);
            return str;
        }
        // 
        char *FindCRLF()
        {
            void *res = memchr(GetReadPosition(), '\n', ReadAbleSize());
            return (char*)res;
        }
        std::string GetLine()
        {
            char *pos = FindCRLF();
            if(pos == nullptr)
            {
                return "";
            }
            uint64_t len = (FindCRLF() - GetReadPosition() + 1); // 加1为了取出结尾的'\n'
            return ReadAsString(len);
        }
        std::string GetLineAndMove()
        {
            char *pos = FindCRLF();
            if(pos == nullptr)
            {
                return "";
            }
            uint64_t len = (FindCRLF() - GetReadPosition() + 1); // 加1为了取出结尾的'\n'
            return ReadAsStringAndMove(len);
        }
        // 清空缓冲区
        void Clear()
        {
            _read_idx = 0;
            _write_idx = 0;
        }
};

#define DEFAULT_LISTEN_SIZE 1024
class Socket
{
    private:
        int _sockfd;
    public:
        Socket(): _sockfd(-1) {}
        Socket(int fd): _sockfd(fd) {}
        ~Socket() { Close(); }
        int Fd() { return _sockfd; }
        // 创建套接字
        bool Create()
        {
            _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
            if(_sockfd < 0)
            {
                ERR_LOG("Create Socket Failed!!");
                return false;
            }
            return true;
        }
        // 绑定端口
        bool Bind(const std::string &ip, uint16_t port)
        {
            struct sockaddr_in addr;
            addr.sin_family = AF_INET;
            addr.sin_port = htons(port);
            addr.sin_addr.s_addr = inet_addr(ip.c_str());
            socklen_t len = sizeof(struct sockaddr_in);
            int ret = bind(_sockfd, (struct sockaddr*)&addr, len);
            if(ret < 0)
            {
                ERR_LOG("Bind Address Failed!! %d", ret);
                return false;
            }
            return true;
        }
        // 开始监听
        bool Listen(int backlog = DEFAULT_LISTEN_SIZE)
        {
            int ret = listen(_sockfd, backlog);
            if(ret < 0)
            {
                ERR_LOG("Socket Listen Failed!!");
                return false;
            }
            return true;
        }
        // 向服务器发起连接
        bool Connect(const std::string &ip, uint16_t port)
        {
            struct sockaddr_in addr;
            addr.sin_family = AF_INET;
            addr.sin_port = htons(port);
            addr.sin_addr.s_addr = inet_addr(ip.c_str());
            socklen_t len = sizeof(struct sockaddr_in);
            int ret = connect(_sockfd, (struct sockaddr*)&addr, len);
            if(ret < 0)
            {
                ERR_LOG("Connect Server Failed!!");
                return false;
            }
            return true;            
        }
        // 获取新连接(返回值为文件描述符)
        int Accept()
        {
            int newfd = accept(_sockfd, nullptr, nullptr);
            if(newfd < 0)
            {
                ERR_LOG("Socket Accept Failed!!");
                return -1;
            }
            return newfd;
        }
        // 接收数据
        ssize_t Recv(void *buf, size_t len, int flag = 0)
        {
            ssize_t ret = recv(_sockfd, buf, len, flag);
            if(ret <= 0)
            {
                // EAGAIN   表示socket接收缓冲区中没有数据了，在非阻塞情况下才会出现这个错误
                // EINTR    表示当前socket的阻塞等待，被信号打断了
                if(errno == EAGAIN || errno == EINTR)
                {
                    return 0; // 表示没有接收到数据，并非出错
                }
                ERR_LOG("Socket Recv Failed!!");
                return -1;
            }
            return ret; // 实际接收到的数据长度
        }
        ssize_t NonblockRecv(void *buf, size_t len)
        {
            return Recv(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT 表示当前接收为非阻塞
        }
        // 发送数据
        ssize_t Send(const void *buf, size_t len, int flag = 0)
        {
            if(len == 0) return 0;
            ssize_t ret = send(_sockfd, buf, len, flag);
            if(ret < 0)
            {
                // EAGAIN   表示socket接收缓冲区中没有数据了，在非阻塞情况下才会出现这个错误
                // EINTR    表示当前socket的阻塞等待，被信号打断了
                if(errno == EAGAIN || errno == EINTR)
                {
                    return 0; // 表示没有发送出去数据，并非出错
                }
                ERR_LOG("Socket Send Failed!!");
                return -1;
            }
            return ret; // 实际发送的数据长度
        }
        ssize_t NonblockSend(void *buf, size_t len)
        {
            return Send(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT 表示当前接收为非阻塞
        }
        // 关闭套接字
        void Close()
        {
            if(_sockfd != -1)
            {
                close(_sockfd);
                _sockfd = -1;
            }
        }
        // 创建服务端连接
        bool CreateServer(uint64_t port, const std::string &ip = "0.0.0.0", bool flag = false)
        {
            // 1.创建套接字
            if(Create() == false) return false;
            // 2.设置非阻塞
            if(flag) Nonblock();
            // 3.绑定地址
            if(Bind(ip, port) == false) return false;
            // 4.开始监听
            if(Listen() == false) return false;
            // 5.启动地址重用
            ReuseAddress();
            return true;
        }
        // 创建客户端连接
        bool CreateClient(uint64_t port, const std::string &ip)
        {
            // 1.创建套接字
            if(Create() == false) return false;
            // 2.指向连接服务器
            if(Connect(ip, port) == false) return false;
            return true;
        }
        // 设置套接字选项---开启地址端口重用
        void ReuseAddress()
        {
            int val = 1;
            setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void*)&val, sizeof(int));
            val = 1;
            setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void*)&val, sizeof(int));
        }
        // 设置套接字阻塞属性---设置为非阻塞
        void Nonblock()
        {
            int flag = fcntl(_sockfd, F_GETFL, 0);
            fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
        }
};

class Poller;
class EventLoop;
class Channel
{
    private:
        int _fd;
        EventLoop *_loop;
        uint32_t _events; // 当前需要监控的事件
        uint32_t _revents; // 当前连接触发的事件
        using EventCallback = std::function<void()>;
        EventCallback _read_callback;   // 可读事件被触发的回调函数
        EventCallback _write_callback;  // 可写事件被触发的回调函数
        EventCallback _error_callback;  // 错误事件被触发的回调函数
        EventCallback _close_callback;  // 连接断开事件被触发的回调函数
        EventCallback _event_callback;  // 任意事件被触发的回调函数
    public:
        Channel(EventLoop *loop, int fd): _fd(fd), _revents(0), _events(0), _loop(loop) {}
        int Fd() { return _fd; }
        uint32_t Event() { return _events; }
        void SetREvents(uint32_t events) { _revents = events; }
        void SetReadCallback(const EventCallback &cb) { _read_callback = cb; }
        void SetWriteCallback(const EventCallback &cb) {_write_callback = cb; }
        void SetErrorCallback(const EventCallback &cb) { _error_callback = cb; }
        void SetCloseCallback(const EventCallback &cb)  { _close_callback = cb; }
        void SetEventCallback(const EventCallback &cb) { _event_callback = cb; }
        // 是否监控了可读
        bool ReadAble() { return (_events & EPOLLIN); }     
        // 是否监控了可写
        bool WriteAble() { return (_events & EPOLLOUT); }   
        // 开启读监控
        void EnableRead()
        {
            _events |= EPOLLIN;
            // 添加至EventLoop的事件监控中
            Update();
        }
        // 开启写监控
        void EnableWrite()
        {
            _events |= EPOLLOUT;
            // 添加至EventLoop的事件监控中
            Update();
        }
        // 关闭读监控
        void DisableRead()
        {
            _events &= ~EPOLLIN;
            // 修改EventLoop的事件监控
            Update();
        }
        // 关闭写监控
        void DisableWrite()
        {
            _events &= ~EPOLLOUT;
            // 修改EventLoop的事件监控
            Update();
        }
        // 关闭所有事件监控
        void DisableAll() { _events = 0; Update(); }
        // 移除监控，需要用到Poller类，在底部实现
        void Remove();
        void Update();
        // 事件处理，一旦触发事件，就调用这个函数      
        void HandleEvent()
        {
            if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP)/*对方关闭连接*/ || (_revents & EPOLLPRI)/*优先数据*/)
            {
                if(_read_callback) _read_callback();
            }
            // 有可能会释放连接的操作，一次只处理一个
            if((_revents & EPOLLOUT))
            {
                if(_write_callback) _write_callback();
            }
            else if((_revents & EPOLLERR))
            {
                // 任何时间都调用的回调函数
                if(_error_callback) _error_callback();
            }
            else if((_revents & EPOLLHUP))
            {
                // 任何时间都调用的回调函数
                if(_close_callback) _close_callback();
            }
            if(_event_callback) _event_callback();
        }
};

#define MAX_EPOLLEVENT 1024
class Poller
{
    private:
        int _epfd;
        struct epoll_event _evs[MAX_EPOLLEVENT];
        std::unordered_map<int, Channel *> _channels;
    private:
        // 判断channel是否已经添加了事件监控
        bool HasChannel(Channel *channel) { return _channels.find(channel->Fd()) != _channels.end(); }
        // 对epoll的直接操作
        void Update(Channel *channel, int op)
        {
            int fd = channel->Fd();
            struct epoll_event ev;
            ev.data.fd = fd;
            ev.events = channel->Event();
            int ret = epoll_ctl(_epfd, op, fd, &ev);
            if(ret < 0)
            {
                ERR_LOG("Epollcrtl Failed!!");
            }
        }
    public:
        Poller()
        {
            _epfd = epoll_create(1);
            if(_epfd < 0)
            {
                ERR_LOG("Epoll Create Failed!!");
                abort(); // 退出程序
            }
        }
        // 添加/更新监控事件
        void UpdateEvent(Channel *channel)
        {
            if(HasChannel(channel))
            {
                return Update(channel, EPOLL_CTL_MOD);
            }
            int fd = channel->Fd();
            _channels[fd] = channel;
            return Update(channel, EPOLL_CTL_ADD);
        }
        // 移除监控事件
        void RemoveEvent(Channel *channel)
        {
            auto it = _channels.find(channel->Fd());
            if(it != _channels.end())
            {
                _channels.erase(it);
            }
            Update(channel, EPOLL_CTL_DEL);
        }
        // 开始监控，返回活跃连接
        void Poll(std::vector<Channel*> *active)
        {
            int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENT, -1);
            if(nfds < 0)
            {
                if(errno == EINTR) return;
                ERR_LOG("Epoll Wait Error:%s\n", strerror(errno));
                abort();
            }
            for(int i = 0; i < nfds; i++)
            {
                auto it = _channels.find(_evs[i].data.fd);
                assert(it != _channels.end());
                it->second->SetREvents(_evs[i].events);
                active->push_back(it->second);
            }
        }
};

using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class TimerTask
{
    private:
        uint64_t _id;               // 定时器任务对象ID
        uint32_t _timeout;          // 超时时间
        bool _canceled;             // 任务是否被取消 true表示被取消
        TaskFunc _task_cb;          // 定时器要执行的定时任务
        ReleaseFunc _release;       // 清空时间轮的对应关系
    public:
        TimerTask(uint64_t id, uint32_t delay, const TaskFunc &cb): _id(id), _timeout(delay), _task_cb(cb), _canceled(false) {}
        ~TimerTask()
        { 
            if(_canceled == false)
                _task_cb(); 
            _release(); 
        }
        void SetReleaseFunc(const ReleaseFunc &cb) { _release = cb; }
        void Cancel() { _canceled = true; }
        uint32_t DelayTime() { return _timeout; }
};

class TimerWheel
{
    private:
        using PtrTask = std::shared_ptr<TimerTask>;
        using WeakTask = std::weak_ptr<TimerTask>;
        uint32_t _capacity; // 最大时间
        uint32_t _tick;     // 当前时间，走到哪释放(执行)到哪
        std::vector<std::vector<PtrTask>> _wheel;
        std::unordered_map<uint64_t, WeakTask> _timers;

        EventLoop *_loop;
        int _timerfd;               // 定时器描述符
        std::unique_ptr<Channel> _timer_channel;
    private:
        void RemoveTimer(uint64_t id) // 从哈希表中移除事件
        {
            auto it = _timers.find(id);
            if(it != _timers.end())
            {
                _timers.erase(it);
            }
        }
        static int CreateTimerFd()
        {
            int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
            if(timerfd < 0)
            {
                ERR_LOG("Timerfd Create Failed!!");
                abort();
            }
            struct itimerspec itime;
            itime.it_value.tv_sec = 1;      // 第一次超时时间为1s
            itime.it_value.tv_nsec = 0;     // 毫秒不需要，设为零
            itime.it_interval.tv_sec = 1;   //后续超时时间也为1s
            itime.it_interval.tv_nsec = 0;
            timerfd_settime(timerfd, 0, &itime, nullptr);
            return timerfd;
        }
        void ReadTimeFd()
        {
            uint64_t times;
            int ret = read(_timerfd, &times, sizeof(times));
            if(ret < 0)
            {
                ERR_LOG("Timefd Read Failed!!");
                abort();
            }
            
        }
        void RunTimerTask()
        {
            _tick = (_tick + 1) % _capacity;
            _wheel[_tick].clear();              // 清空数组自然会释放所有指针
        }
        // 到时间了
        void OnTime()
        {
            ReadTimeFd();
            RunTimerTask();
        }
        void TaskAddInLoop(uint64_t id, uint32_t delay, const TaskFunc &cb)   // 新加任务
        {
            PtrTask pt(new TimerTask(id, delay, cb));
            pt->SetReleaseFunc(std::bind(&TimerWheel::RemoveTimer, this, id));
            int pos = (_tick + delay) % _capacity;
            _wheel[pos].push_back(pt);
            _timers[id] = WeakTask(pt);
        }
        void TaskRefreshInLoop(uint64_t id)                                   
        {
            auto it = _timers.find(id);
            if(it == _timers.end())
                return; // 没找到该ID对应的任务，直接返回
            PtrTask pt = it->second.lock();
            int delay = pt->DelayTime();
            int pos = (_tick + delay) % _capacity;
            _wheel[pos].push_back(pt);
        }
        void CancelTaskInLoop(uint64_t id)
        {
            auto it = _timers.find(id);
            if(it == _timers.end())
                return; // 没找到该ID对应的任务，直接返回
            PtrTask pt = it->second.lock();
            // Question pt是nullptr，但it->second使用gdb可以正常读取数值
            if(pt) pt->Cancel();
        }
    public:
        TimerWheel(EventLoop *loop): _capacity(60), _tick(0), _wheel(_capacity), _timerfd(CreateTimerFd()), 
                                   _loop(loop), _timer_channel(new Channel(loop, _timerfd)) 
        {
            _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime, this));
            _timer_channel->EnableRead();
        }
        // 三个函数需要调用RunInLoop函数，所以放在文件结尾实现
        void TaskAdd(uint64_t id, uint32_t delay, const TaskFunc &cb);
        void TaskRefresh(uint64_t id);   // 刷新/延迟任务
        void CancelTask(uint64_t id);
        // 存在线程安全问题，只能在对应的EventLoop中被调用
        bool HasTimer(uint64_t id) { return _timers.find(id) != _timers.end(); }
};

class EventLoop
{
    private:
        using Functor = std::function<void()>;
        std::thread::id _thread_id; // 线程ID，不在线程内执行的加入任务队列
        int _event_fd;              // 为了唤醒阻塞等待监控的
        std::unique_ptr<Channel> _event_channel;    
        Poller _poller;             // 进行所有描述符的事件监控
        std::vector<Functor> _tasks;// 任务队列
        std::mutex _tasks_mutex;    // 任务队列的锁
        TimerWheel _timer_wheel;    // 定时器模块
    public:
        // 执行任务池中的所有任务
        void RunAllTask()
        {
            std::vector<Functor> functor;
            {
                std::unique_lock<std::mutex> Lock(_tasks_mutex);
                _tasks.swap(functor);
            }
            for(auto &f : functor)
            {
                f();
            }
        }
        static int CreateEventFd()
        {
            int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
            if(efd < 0)
            {
                ERR_LOG("Eventfd Create Failed!!");
                abort();
            }
            return efd;
        }
        void ReadEventFd()
        {
            uint64_t res = 0;
            bool ret = read(_event_fd, &res, sizeof(res));
            if(ret < 0)
            {
                // EINTR表示被信号打断      EAGAIN表示无数据可读
                if(errno == EINTR || errno == EAGAIN) return;
                ERR_LOG("Eventfd Read Failed!!");
                abort();
            }
            return;
        }
        void WeakUpEventFd()
        {
            uint64_t val = 1;
            bool ret = write(_event_fd, &val, sizeof(val));
            if(ret < 0)
            {
                if(errno == EINTR) return;
                ERR_LOG("Eventfd Weak Up Failed!!");
                abort();
            }
            return;
        }
    public:
        EventLoop(): _thread_id(std::this_thread::get_id()),
                    _event_fd(CreateEventFd()),
                    _event_channel(new Channel(this, _event_fd)),
                    _timer_wheel(this)
        {
            // 给eventfd添加可读事件回调函数，读取eventfd事件通知次数
            _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd, this));
            _event_channel->EnableRead();
        }
        // 判断该任务是否能在线程内执行，不能执行压入任务队列
        void RunInLoop(const Functor &cb)
        {
            if(IsInLoop())
            {
                return cb();
            }
            return QueueInLoop(cb);
        }
        void AssertInLoop()
        {
            assert(_thread_id == std::this_thread::get_id());
        }
        // 将任务压入任务池
        void QueueInLoop(const Functor &cb)
        {
            {
                std::unique_lock<std::mutex> Lock(_tasks_mutex);
                _tasks.push_back(cb);
            }
            // 唤醒有可能因为没有事件就绪，而导致的epoll阻塞
            WeakUpEventFd();
        }
        // 判断是否在同一线程
        bool IsInLoop() {  return std::this_thread::get_id() == _thread_id; }
        // 添加/修改描述符的事件监控
        void UpdateEvent(Channel *channel) { return _poller.UpdateEvent(channel); }
        void RemoveEvent(Channel *channel) { return _poller.RemoveEvent(channel); }
        void Start()
        {
            while(1)
            {
                std::vector<Channel *> actives;
                // 1.事件监控
                _poller.Poll(&actives);
                // 2.事件处理
                for(auto &channel : actives)
                {
                    channel->HandleEvent();
                }
                // 3.执行任务
                RunAllTask();
            }
        }
        void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb) { return _timer_wheel.TaskAdd(id, delay, cb); }
        void TimerRefresh(uint64_t id) { return _timer_wheel.TaskRefresh(id); }
        void TimerCancel(uint64_t id) { return _timer_wheel.CancelTask(id); }
        bool HasTimer(uint64_t id) { return _timer_wheel.HasTimer(id); }
};

class Connection;
typedef enum
{
    DISCONNECTED,/*连接关闭状态*/
    CONNECTING,/*连接建立成功，待处理阶段*/
    CONNECTED,/*连接建立完成，可以通信*/
    DISCONNECTING /*待关闭*/
} ConnStatu;
using PtrConnection = std::shared_ptr<Connection>;
class Connection: public std::enable_shared_from_this<Connection>
{
    private:
        uint64_t _conn_id;              // 连接的唯一ID
        // uint64_t _timer_id;  // 定时器ID，必须唯一，此处简化操作，使用_conn_id代替
        int _sockfd;                    // 连接关联的文件描述符
        bool _enable_inactive_release;  // 连接是否开启非活跃销毁，默认为false——不启动
        EventLoop* _loop;    // 连接关联的eventloop
        ConnStatu _statu;               // 连接当前状态
        Socket _socket;                 // 套接字操作管理
        Channel _channel;               // 连接的时间管理
        Buffer _in_buffer;              // 输入缓冲区
        Buffer _out_buffer;             // 输出缓冲区——存放要发送的数据
        Any _context;                   // 请求的接受处理上下文

        using ConnectedCallBack = std::function<void(const PtrConnection&)>;
        using MessageCallBack = std::function<void(const PtrConnection&, Buffer*)>;
        using CloseCallBack = std::function<void(const PtrConnection&)>;
        using AnyEventCallBack = std::function<void(const PtrConnection&)>;
        ConnectedCallBack _connected_callback;
        MessageCallBack _message_callback;
        CloseCallBack _closed_callback;
        AnyEventCallBack _event_callback;

        CloseCallBack _server_closed_callback;
    private:
        // 五个Channel的事件回调函数
        void HandleRead()
        {
            // 1.接收数据放入缓冲区
            char buf[65536] = {0};
            ssize_t ret = _socket.NonblockRecv(buf, 65535);
            if(ret < 0)
            {
                ERR_LOG("Connection Read Error!!");
                return ShutdownInLoop();
            }
            _in_buffer.WriteAndMove(buf, ret);
            // 2.业务处理
            if(_in_buffer.ReadAbleSize() > 0)
            {
                return _message_callback(shared_from_this(), &_in_buffer);
            }
        }
        void HandleWrite()
        {
            ssize_t ret = _socket.NonblockSend(_out_buffer.GetReadPosition(), _out_buffer.ReadAbleSize());
            if(ret < 0)
            {
                ERR_LOG("Connection Write Error!!");
                if(_in_buffer.ReadAbleSize() > 0)
                {
                    if(_message_callback) _message_callback(shared_from_this(), &_in_buffer);
                }
                return Release();
            }
            _out_buffer.MoveReadOffset(ret);
            if(_out_buffer.ReadAbleSize() == 0)
            {
                _channel.DisableWrite();    // 没有数据待发送了，关闭写事件监控
                if(_statu == DISCONNECTING)
                {
                    return Release();
                }
            }
        }
        // 描述符触发挂断事件
        void HandleClose()
        {
            if(_in_buffer.ReadAbleSize() > 0)
            {
                if(_message_callback) _message_callback(shared_from_this(), &_in_buffer);
            }
            return Release();
        }
        // 描述符触发出错事件
        void HandleError()
        {
            return HandleClose();
        }
        // 描述符触发任意事件
        void HandleEvent()
        {
            // 1.刷新活跃度
            if(_enable_inactive_release == true)
            {
                _loop->TimerRefresh(_conn_id);
            }
            // 2.调用设置的回调函数
            if(_event_callback)
            {
                _event_callback(shared_from_this());
            }
        }
        // 连接获取之后，进行回调函数等的设置
        void EstablishedInLoop()
        {
            assert(_statu == CONNECTING);
            // 启动读事件监控
            _channel.EnableRead();
            // 调用回调函数
            if(_connected_callback)
            {
                _connected_callback(shared_from_this());
            }
            // 修改连接状态
            _statu = CONNECTED;
        }
        void ReleaseInLoop()
        {
            // 修改连接状态
            _statu = DISCONNECTED;
            // 移除事件监控
            _channel.Remove();
            // 关闭描述符
            _socket.Close();
            // 如果还有定时销毁，则取消
            if(_loop->HasTimer(_conn_id)) _loop->TimerCancel(_conn_id);
            // 调用关闭回调函数
            if(_closed_callback) _closed_callback(shared_from_this());
            // 移除服务器内部管理的连接信息
            if(_server_closed_callback) _server_closed_callback(shared_from_this());
        }
        void SendInLoop(Buffer buf)
        {
            if(_statu == DISCONNECTED) return;
            // 将数据放到缓冲区
            _out_buffer.WriteFromBufferAndMove(buf);
            // 启动可写事件监控
            if(_channel.WriteAble() == false)
            {
                _channel.EnableWrite();
            }
        }
        void ShutdownInLoop()
        {
            _statu = DISCONNECTING; // 设置连接为半关闭状态
            if(_out_buffer.ReadAbleSize())
            {
                if(_channel.WriteAble() == false) _channel.EnableWrite();
            }
            if(_in_buffer.ReadAbleSize() > 0)
            {
                if(_message_callback) _message_callback(shared_from_this(), &_in_buffer);
            }
            if(_in_buffer.ReadAbleSize() == 0)
            {
                Release();
            }
        }
        void EnableInactiveReleaseInLoop(int sec)
        {
            // 将判断标志位置为true
            _enable_inactive_release = true;
            // 如果存在定时销毁任务，则刷新
            if(_loop->HasTimer(_conn_id))
            {
                return _loop->TimerRefresh(_conn_id);
            }
            // 如果不存在，则新增
            _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::ReleaseInLoop, this));
        }
        void CancelInactiveReleaseInLoop()
        {
            _enable_inactive_release = false;
            if(_loop->HasTimer(_conn_id))
            {   
                _loop->TimerCancel(_conn_id);
            }
        }
        void UpgradeInLoop(const Any &context, const ConnectedCallBack &conn, const MessageCallBack &msg, 
                           const CloseCallBack &closed, const AnyEventCallBack &event)
        {
            _context = context;
            _connected_callback = conn;
            _message_callback = msg;
            _closed_callback = closed;
            _event_callback = event;
        }
    public:
        Connection(EventLoop *loop, uint64_t conn_id, int sockfd):
        _loop(loop), _conn_id(conn_id), _sockfd(sockfd), _enable_inactive_release(false),
        _statu(CONNECTING), _socket(_sockfd), _channel(loop, sockfd)
        {
            _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
            _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
            _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
            _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
            _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        }
        ~Connection() { DBG_LOG("Release Connection: %p", this); }
        int Fd() { return _sockfd; }                                // 获取文件描述符
        int Id() { return _conn_id; }                               // 获取连接ID
        bool Connected() { return (_statu == CONNECTED); }          // 是否处于CONNECTED状态
        void SetContext(const Any &context) { _context = context; } // 设置上下文
        Any *GetContext() { return &_context; }                  // 获取上下文
        void SetConnectedCallBack(const ConnectedCallBack &cb) { _connected_callback = cb; }
        void SetMessageCallBack(const MessageCallBack &cb) { _message_callback = cb; }
        void SetClosedCallBack(const CloseCallBack &cb) { _closed_callback = cb; }
        void SetAnyEventCallBack(const AnyEventCallBack &cb) { _event_callback = cb; }
        void SetSrvClosedCallBack(const AnyEventCallBack &cb) { _server_closed_callback = cb; }
        // 连接就绪后设置Channel回调设置，启动读监控等
        void Established()
        {
            _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
        }       
        // 发送数据  
        void Send(const char *data, size_t len)
        {
            Buffer buf;
            buf.WriteAndMove(data, len);    
            _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, buf));
        }
        // 释放线程所持有的资源
        void Release()
        {
            _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this));
        }
        // 提供给组件使用者的关闭操作(发送缓冲区未清空时处于待关闭状态而不是直接关闭)
        void Shutdown()
        {
            _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
        }
        // 开启超时销毁，需传入超时时间
        void EnableInactiveRelease(int sec)
        {
            _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
        }
        // 取消超时销毁
        void CancelInactiveRelease()
        {
            _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
        }
        // 更新协议——重置上下文和阶段性处理函数——这个接口必须在EventLoop线程中被立即执行
        void Upgrade(const Any &context,const ConnectedCallBack &conn, const MessageCallBack &msg, 
                     const CloseCallBack &closed, const AnyEventCallBack &event)
        {
            _loop->AssertInLoop();
            _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
        }
};

class Acceptor
{
    private:
        Socket _socket;     // 创建套接字
        EventLoop *_loop;   // 事件监控
        Channel _channel;   // 监听套接字管理

        using AcceptCallBack = std::function<void(int)>;
        AcceptCallBack _accept_callback;
    private:
        // 监听套接字的读事件回调
        void HandleRead()
        {
            int newfd = _socket.Accept();
            if(newfd < 0) return;
            if(_accept_callback) _accept_callback(newfd);
        }
        int CreateServer(uint64_t port, const std::string &ip = "0.0.0.0", bool flag = false)
        {
            bool ret = _socket.CreateServer(port, ip, flag);
            assert(ret == true);
            return _socket.Fd();
        }
    public:
        Acceptor(EventLoop *loop, int port): _loop(loop), _socket(CreateServer(port)), _channel(loop, _socket.Fd())
        {
            _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
        }
        void SetAcceptCallBack(const AcceptCallBack &cb) { _accept_callback = cb; }
        void Listen() { _channel.EnableRead(); }

};

class LoopThread
{
    private:
        std::mutex _mutex;
        std::condition_variable _cond;
        std::thread _thread;    // EventLoop对应的线程
        EventLoop *_loop;       // 需要在线程内实例化
    private:
        // 线程的入口函数，实例化EventLoop对象，唤醒_cond上阻塞的线程
        void ThreadEntry()
        {
            EventLoop loop;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _loop = &loop;
                _cond.notify_all();
            }
            loop.Start();
        }
    public:
        LoopThread(): _loop(nullptr), _thread(std::thread(&LoopThread::ThreadEntry, this)) {}
        // 返回当前线程关联的loop
        EventLoop *GetLoop()
        {
            EventLoop *loop = nullptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _cond.wait(lock, [&](){ return _loop != nullptr; });
                loop = _loop;
            }
            return loop;
        }
};

class LoopThreadPool
{
    private:
        int _thread_count;                  // 从属线程数量
        int _next_loop_idx;                 // 采用RR轮转(轮询调度)，记录下标
        EventLoop *_baseloop;               // 主EventLoop，线程池数量为0则挂载在主线程下
        std::vector<LoopThread*> _threads;  // 线程池
        std::vector<EventLoop*> _loops;     // 线程对应的EvnetLoop
    public:
        LoopThreadPool(EventLoop *baseloop): _thread_count(0), _next_loop_idx(0), _baseloop(baseloop) {}
        // 设置线程数量
        void SetThreadCount(int count)
        {
            _thread_count = count;
        }
        // 创建所有从属线程
        void Create()
        {
            if(_thread_count > 0)
            {
                _threads.resize(_thread_count);
                _loops.resize(_thread_count);
                for(int i = 0; i < _thread_count; i++)
                {
                    _threads[i] = new LoopThread();
                    _loops[i] = _threads[i]->GetLoop();
                }
            }
        }
        // 获取下一个Eventloop
        EventLoop *NextLoop()
        {
            if(_thread_count == 0)
            {
                return _baseloop;
            }
            _next_loop_idx = (_next_loop_idx + 1) % _thread_count;
            return _loops[_next_loop_idx];
        }
};

class TcpServer
{
    private:
        uint64_t _next_id;              // 自动增长的连接ID
        int _port;                      // 服务器端口
        int _timeout;                   // 超时销毁时长——多久无通信就属于非活跃连接
        bool _enable_inactive_release;  // 是否启动非活跃连接超时销毁
        EventLoop _baseloop;           // 主线程的loop
        Acceptor _acceptor;             // 监听套接字管理
        LoopThreadPool _pool;           // 从属线程池
        std::unordered_map<uint64_t, PtrConnection> _conns;

        using Functor = std::function<void()>;
        using ConnectedCallBack = std::function<void(const PtrConnection&)>;
        using MessageCallBack = std::function<void(const PtrConnection&, Buffer*)>;
        using CloseCallBack = std::function<void(const PtrConnection&)>;
        using AnyEventCallBack = std::function<void(const PtrConnection&)>;
        ConnectedCallBack _connected_callback;
        MessageCallBack _message_callback;
        CloseCallBack _closed_callback;
        AnyEventCallBack _event_callback;
    private:
        void RemoveConnectionInLoop(const PtrConnection &conn)
        {
            uint64_t id = conn->Id();
            auto it = _conns.find(id);
            if(it != _conns.end()) _conns.erase(it);
        }
        // 从管理的_conns中移除连接信息
        void RemoveConnection(const PtrConnection &conn)
        {
            _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn)); 
        }
        void NewConnection(int fd)
        {
            _next_id++;
            PtrConnection conn(new Connection(_pool.NextLoop(), _next_id, fd));
            conn->SetMessageCallBack(_message_callback);
            conn->SetClosedCallBack(_closed_callback);
            conn->SetConnectedCallBack(_connected_callback);
            conn->SetAnyEventCallBack(_event_callback);
            conn->SetSrvClosedCallBack(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
            // 非活跃连接释放任务
            if(_enable_inactive_release) conn->EnableInactiveRelease(_timeout);
            conn->Established();
            _conns.insert({_next_id, conn});
            DBG_LOG("Accept A New Connection");
        }
        void RunAfterInLoop(const Functor &task, int delay)
        {
            _next_id++;
            _baseloop.TimerAdd(_next_id, delay, task);
        }
    public:
        TcpServer(int port):
          _next_id(0), 
          _port(port), 
          _enable_inactive_release(false), 
          _acceptor(&_baseloop, port), 
          _pool(&_baseloop)
        {
            _acceptor.SetAcceptCallBack(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
            _acceptor.Listen(); // 将监听套接字挂载到baseloop上
        }
        void SetThreadCount(int count) { return _pool.SetThreadCount(count); }
        void SetConnectedCallBack(const ConnectedCallBack &cb) { _connected_callback = cb; }
        void SetMessageCallBack(const MessageCallBack &cb) { _message_callback = cb; }
        void SetClosedCallBack(const CloseCallBack &cb) { _closed_callback = cb; }
        void SetAnyEventCallBack(const AnyEventCallBack &cb) { _event_callback = cb; }
        void EnableInactiveRelease(int timeout)
        {
            _timeout = timeout;
            _enable_inactive_release = true;
        }
        void RunAfter(const Functor &task, int delay)
        {
            _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));
        }
        void Start()
        {
            _pool.Create();     // 创建从属线程
            _baseloop.Start();
        }

};

void Channel::Remove() { return _loop->RemoveEvent(this); }
void Channel::Update() { return _loop->UpdateEvent(this); }
void TimerWheel::TaskAdd(uint64_t id, uint32_t delay, const TaskFunc &cb)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TaskAddInLoop, this, id, delay, cb));
}
void TimerWheel::TaskRefresh(uint64_t id)   // 刷新/延迟任务
{
    _loop->RunInLoop(std::bind(&TimerWheel::TaskRefreshInLoop, this, id));
}
void TimerWheel::CancelTask(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::CancelTaskInLoop, this, id));
}

class NetWork
{
    public:
        NetWork()
        {
            DBG_LOG("SIGPIPE Init");
            signal(SIGPIPE, SIG_IGN);
        }
};
static NetWork nw;
#endif